package Exercises;

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Instruction：基于阻塞队列实现生产者消费者模型
 * Author：@author MaLi
 */
public class Exercise03 {
    private static AtomicInteger atomicInteger = new AtomicInteger(0);

    public static void main(String[] args) {
        DataBuffer dataBuffer = new DataBuffer();
        ExecutorService pool = Executors.newFixedThreadPool(8);
        //提交三个生产者
        for (int i = 0; i < 3; i++) {
            pool.submit(new Producer(dataBuffer));
        }
        //提交五个消费者
        for (int i = 0; i < 5; i++) {
            pool.submit(new Consumer(dataBuffer));
        }
    }

    //创建DataBuffer, 用于存储消息
    private static class DataBuffer {
        private ArrayBlockingQueue<String> container = new ArrayBlockingQueue<>(10);

        public void put(String element) {
            try {
                container.put(element);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        public String fetch() {
            try {
                return container.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private static class Producer implements Runnable {
        private DataBuffer dataBuffer;

        public Producer(DataBuffer dataBuffer) {
            this.dataBuffer = dataBuffer;
        }

        @Override
        public void run() {
            String tName = Thread.currentThread().getName();
            while (true) {
                dataBuffer.put(tName + " --" + atomicInteger.incrementAndGet());
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private static class Consumer implements Runnable {
        private DataBuffer dataBuffer;

        public Consumer(DataBuffer dataBuffer) {
            this.dataBuffer = dataBuffer;
        }

        @Override
        public void run() {
            String tName = Thread.currentThread().getName();
            while (true) {
                String data = dataBuffer.fetch();
                System.out.println(tName + " fetch: " + data);
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}
